package hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;


/**
 * @author Judy
 * @data 2022/3/12 16:15
 */
public class UserLogFormHBase {
    // 放置初始化变量
    public static Configuration config;

    public void UserLogFromHBase() {

    }

    // 放置初始化方法
    public void initConfig() {
        config = HBaseConfiguration.create();// 配置
        config.set("hbase.zookeeper.quorum", "192.168.56.69");// zookeeper地址
        config.set("hbase.zookeeper.property.clientPort", "2181");// zookeeper端口
        config.setInt("hbase.rpc.timeout", 20000);
        config.setInt("hbase.client.operation.timeout", 20000);
        config.setInt("hbase.client.scanner.timeout.period", 20000);
        config.set("hbase.client.ipc.pool.type", "RoundRobinPool");
    }

    /**
     * 获取指定RowKey对应的所有列的值，并通过控制台打印出来
     *
     * @param rowkey
     * @throws IOException
     */
    public void getDataDetailByRowKey(String rowkey) {
        initConfig();
        // 创建HBase的连接
        try {
            Connection connection = ConnectionFactory.createConnection(config);
            // 获取User_Interest_Log的表
            Table table = connection.getTable(TableName.valueOf("Student"));
            // 根据传入的RowKey获取数据
            Get get = new Get(Bytes.toBytes(rowkey));
            Result result = table.get(get);

            // 遍历返回行的所有列以及数据
            for (Cell cell : result.rawCells()) {
                System.out.print("列族=>");
                System.out.print(new String(CellUtil.cloneFamily(cell)));
                System.out.print(":");
                System.out.print(new String(CellUtil.cloneQualifier(cell)));
                System.out.print("=");
                System.out.print(new String(CellUtil.cloneValue(cell)));
                System.out.print(";");
                System.out.print("Timestamp=");
                System.out.println(cell.getTimestamp());
            }

        }catch (Exception e){
            System.out.println(e);
            e.printStackTrace();
        }
    }

    /**
     * 查询当前表中说有的用户兴趣日志（在真实业务系统中，不允许不带条件的全表扫描）
     *
     * @return 经过处理的用户兴趣日志
     * @throws IOException
     */
    public List<String> getUserLogList(String tableName, String[] columnFamilies)  {
        initConfig();
        List<String> userlogList = new ArrayList();
        // 从HBase中获取表的所有数据，并返回格式化后的结果
        // 判断如果传入的参数不合规，直接返回空值
        if (columnFamilies == null || tableName.isEmpty() || columnFamilies == null || columnFamilies.length == 0) {
            return userlogList;
        }
        // 创建HBase的连接
        try {
            Connection connection = ConnectionFactory.createConnection(config);
            System.out.println("创建的链接为："+connection);
            // 获取参数指定的表
            Table table = connection.getTable(TableName.valueOf(tableName));
            // 创建Scan对象
            Scan scan = new Scan();
            // 设置scan命令每次获取数据的条数
            scan.setBatch(100);

            // 添加参数指定的列族到Scan对象
            for (String cf : columnFamilies) {
                scan.addFamily(Bytes.toBytes(cf));
            }

            // 获取扫描的返回结果
            ResultScanner resultScanner = table.getScanner(scan);
            Result rs = resultScanner.next();

            // 遍历查询到的所有行，并对行内的Cell进行遍历，获取列名和对应的值
            for (; rs != null; rs = resultScanner.next()) {
                for (Cell cell : rs.listCells()) {
                    //RowKey
                    String rowkey = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()) + ":";
                    // 列族
                    String colFamily = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()) + ":";
                    //列
                    String column = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()) + "=";
                    //值
                    String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    userlogList.add(rowkey + colFamily + column + value);
                    System.out.println(rowkey +"||"+ colFamily +"||"+ column +"||"+ value);
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return userlogList;
    }

}
